/*
 * 文件名：StreamDataSenderHandler.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月13日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.netty.handler;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.net.SocketAddress;
import java.util.concurrent.CopyOnWriteArrayList;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.net.NettyUtil;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUnit;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description 数据分发被动端口监听服务类
 * @author [yuanze]
 * @version [版本号,2014年11月13日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
@Sharable
public class StreamDataPassivSenderHandler extends SimpleChannelInboundHandler<String>
{
	private static Logger logger = LoggerFactory.getLogger(StreamDataPassivSenderHandler.class);

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelInactive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{
		String port = NettyUtil.getLocalPort(ctx);
		CopyOnWriteArrayList<RealStreamUnit> unitList = RealStreamTool.DATAPORT_UNIT_CACHE.get(Integer.valueOf(port));
		for (RealStreamUnit unit : unitList)
		{
			RealStreamTool.DISPATCH_CAHCE.get(unit.getOnlyUnitCode()).rmChannel(ctx.channel().remoteAddress().toString(), ctx.channel());
		}
		logger.info("监测到remote[" + ctx.channel().remoteAddress() + "]关闭,处理缓存信息");
	}

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelActive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		// 缓存被动端口的channel
		String port = NettyUtil.getLocalPort(ctx);
		CopyOnWriteArrayList<RealStreamUnit> unitList = RealStreamTool.DATAPORT_UNIT_CACHE.get(Integer.valueOf(port));
		for (RealStreamUnit unit : unitList)
		{
			RealStreamTool.DISPATCH_CAHCE.get(unit.getOnlyUnitCode()).addChannel(ctx.channel().remoteAddress().toString(), ctx.channel());
			logger.info("received remoteAdd [" + ctx.channel().remoteAddress() + "] 发送消息类型[" + unit.getOnlyUnitCode() + "] request.........");
		}
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception
	{
		SocketAddress clientAddr = ctx.channel().remoteAddress();
		logger.info("receive data transmission request from ip [" + clientAddr.toString() + "] , ready to send data ............");
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
	{
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
	{
		logger.error("receive message exception happened ,caused by " + cause.getMessage());
		String port = NettyUtil.getLocalPort(ctx);
		CopyOnWriteArrayList<RealStreamUnit> unitList = RealStreamTool.DATAPORT_UNIT_CACHE.get(Integer.valueOf(port));
		for (RealStreamUnit unit : unitList)
		{
			RealStreamTool.DISPATCH_CAHCE.get(unit.getOnlyUnitCode()).rmChannel(ctx.channel().remoteAddress().toString(), ctx.channel());
		}
		ctx.close();
	}

}
